package com.example.function.integration.service.impl;

import com.alibaba.fastjson2.JSONObject;
import com.example.function.integration.entity.HuanengImageDataEntity;
import com.example.function.integration.service.AbstractCommonDataService;
import com.example.function.integration.service.ImageFileDownloadService;
import com.example.function.integration.sftp.CommonSftpComponent;
import com.example.function.integration.utils.SftpUtil;
import com.google.common.collect.Lists;
import org.apache.commons.lang3.ObjectUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.io.ByteArrayInputStream;
import java.io.InputStream;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Future;

import static com.example.function.integration.utils.SftpUtil.existStaticFile;


/**
 * @description：影像件下载服务接口实现类
 * @className：ImageFileDownloadServiceImpl
 * @author：ext.fengcheng
 * @date：2023/9/7 10:26
 */
@Service
public class ImageFileDownloadServiceImpl extends AbstractCommonDataService implements ImageFileDownloadService {

    public ConcurrentHashMap<String, SftpUtil> map = new ConcurrentHashMap<>();

    protected static Logger logger = LoggerFactory.getLogger(ImageFileDownloadServiceImpl.class);

    /**
     * @Description：下载文件并且上传文件到SFTP
     * @Author： ext.fengcheng
     * @Date：2023/9/19 9:55
     * @tenantId, 租户id
     * @queryStartId, 查询开始id
     * @queryNum, 每次查询条数
     * @batchNum, 根据每次查询结果，每个批次处理多少条
     * @isDelDirectory 是否进行测试
     * @Return： boolean
     */
    @Override
    public boolean downloadFilesAndUploadFilesToSftp(String tenantId, Long queryStartId, Long queryNum, Long batchNum, Boolean isTest) {
        if (ObjectUtils.isEmpty(queryStartId) || QUERY_START_ID.equals(queryStartId)) {
            if (ObjectUtils.isNotEmpty(localQueue)) {
                int minStartId = findMin(localQueue);
                queryStartId = (long) minStartId;
            } else {
                queryStartId = QUERY_START_ID;
            }
        }
        if (ObjectUtils.isEmpty(queryNum)) queryNum = QUERY_NUM;
        if (ObjectUtils.isEmpty(batchNum)) batchNum = BATCH_NUM;
        if (ObjectUtils.isNotEmpty(isTest)) IS_TEST = isTest;

        Map<String, List<HuanengImageDataEntity>> fileKeyMap = new ConcurrentHashMap<>();
        try {
            while (true) {
                logger.info("借据数据查询，开始ID:{}，每次查询数量：{}，SFTP链接集合数量：{}", queryStartId, queryNum, map.size());
                if (!map.isEmpty()) {
                    logger.info("进行下一次文件上传，关闭SFTP链接");
                    map.forEach((key, value) -> value.disconnect());
                }
                // 每次查询出1000条（条数可配置），借据编号和放款编号
                List<HuanengImageDataEntity> huanengIouDataList = iouRecordExtendMapper.getHuanengIouData(tenantId, queryStartId, queryNum);
                if (CollectionUtils.isEmpty(huanengIouDataList)) {
                    logger.info("未查询到借据数据,开始处理最后一批次数据！！！");
                    if (!CollectionUtils.isEmpty(fileKeyMap)) {
                        splitBigMapToSmailMap(batchNum, fileKeyMap);
                        fileKeyMap = new ConcurrentHashMap<>();
                    }
                    break;
                }
                // 是否测试开关
                if (IS_TEST) queryNum = 0L;
                queryStartId = huanengIouDataList.get(huanengIouDataList.size() - 1).getId();
                // 通过借据编号和放款编号获取文件下载key，然后组装借据维度的map集合
                List<HuanengImageDataEntity> batchList = Lists.newArrayList();
                for (int i = 0; i < huanengIouDataList.size(); i++) {
                    batchList.add(huanengIouDataList.get(i));
                    // 每个批次集合中存满200个元素(可配置)，就开启一个线程去查询数据库获取文件下载key
                    if (batchList.size() == batchNum) {
                        getFileDownloadKeyList(batchList, fileKeyMap);
                        batchList = Lists.newArrayList();
                    }
                    if (i == huanengIouDataList.size() - 1 && !batchList.isEmpty()) {
                        getFileDownloadKeyList(batchList, fileKeyMap);
                        batchList = Lists.newArrayList();
                    }
                }
                // 指定map中有1千个元素（可配置），便开始拉取一次影像件
                if (!fileKeyMap.isEmpty()) {
                    splitBigMapToSmailMap(batchNum, fileKeyMap);
                    fileKeyMap = new ConcurrentHashMap<>();
                }
            }
        } catch (Exception e) {
            logger.info("异常结束，借据查询开始id：{},失败借据id集合：{}", queryStartId, JSONObject.toJSONString(localQueue));
            logger.error("下载文件并且上传文件到SFTP，查询借据数据、以及借据维度的影像件数据,发生异常：{}", e.getMessage(), e);
            throw new RuntimeException(e);
        }
        logger.info("借据数据查询结束，影像件数据查询结束，影像件拉取结束，影像件上传结束,全部逻辑处理结束，失败借据id集合：{}", JSONObject.toJSONString(localQueue));
        return true;
    }

    /**
     * @Description：以借据为维度获取当前借据下的所有文件下载key
     * @Author： ext.fengcheng
     * @Date：2023/9/19 10:45
     * @Param： [entity, fileKeyMap]
     * @Return： void
     */
    private void getFileDownloadKeyList(List<HuanengImageDataEntity> batchList, Map<String, List<HuanengImageDataEntity>> fileKeyMap) {
        if (CollectionUtils.isEmpty(batchList)) {
            return;
        }
        batchList.forEach(entity -> {
            String businessNo = "'" + entity.getLendCode() + "','" + entity.getCreditApplyNo() + "'";
            List<HuanengImageDataEntity> huanengImageDataList = archiveInfoExtendMapper.getHuanengImageData(entity.getTenantId(), businessNo);
            if (!CollectionUtils.isEmpty(huanengImageDataList)) {
                fileKeyMap.put(entity.getIouCode() + "_" + entity.getId(), huanengImageDataList);
            }
        });
    }

    /**
     * @Description：将一个大的Map集合拆分成List集合，将List集合拆分成指定数量的小List集合后，开启一个线程去做拉取和上传操作
     * @Author： ext.fengcheng
     * @Date：2023/10/11 16:44
     * @Param： [batchNum, fileKeyMap]
     * @Return： void
     */
    public void splitBigMapToSmailMap(Long batchNum, Map<String, List<HuanengImageDataEntity>> fileKeyMap) {
        if (!fileKeyMap.isEmpty()) {
            List<Future<Boolean>> futures = new ArrayList<>();
            List<Map<String, List<HuanengImageDataEntity>>> fileKeyMapList = splitMap(fileKeyMap);
            //分批并发处理，提高性能
            int beginIndex = 0;
            int endIndex = 0;
            while (fileKeyMapList.size() > beginIndex) {
                endIndex = beginIndex + Integer.parseInt(batchNum.toString());
                if (fileKeyMapList.size() < endIndex) {
                    endIndex = fileKeyMapList.size();
                }
                List<Map<String, List<HuanengImageDataEntity>>> batchList = fileKeyMapList.subList(beginIndex, endIndex);
                Future<Boolean> future = fileUploadExecutor.submit(() -> {
                    SftpUtil sftpObj = commonSftpComponent.initSftpConnection();
                    fromOssDownloadFileAndUpload(sftpObj, batchList);
                    logger.info("往全局缓存Map中放入SFTP链接，当前线程名称：{}", Thread.currentThread().getName());
                    map.put(Thread.currentThread().getName(), sftpObj);
                    return true;
                });
                futures.add(future);
                beginIndex = endIndex;
            }
            for (Future<Boolean> f : futures) {
                try {
                    Boolean result = f.get();
                    if (!result) {
                        logger.error("等待多线程上传文件处理结果，处理失败，失败借据id集合：{}", JSONObject.toJSONString(localQueue));
                        throw new RuntimeException("等待多线程上传文件处理结果，处理失败");
                    }
                    logger.info("通过影像件url地址从OSS下载文件，并上传文件到SFTP服务器，处理结束，失败借据id集合：{}", JSONObject.toJSONString(localQueue));
                } catch (Exception e) {
                    logger.error("以借据为维度获取当前借据下的所有文件下载key,获取多线程处理结果，，失败借据id集合：{}", JSONObject.toJSONString(localQueue), e);
                    throw new RuntimeException(e);
                }
            }
        }
    }

    /**
     * @Description：开始从OSS中下载文件，并且上传文件到SFTP,下载和上传操作处于同步流程中
     * @Author： ext.fengcheng
     * @Date：2023/10/11 16:52
     * @Param： [fileKeyMap]
     * @Return： void
     */
    public void fromOssDownloadFileAndUpload(SftpUtil sftpUtil, List<Map<String, List<HuanengImageDataEntity>>> batchList) {
        logger.info("fromOssDownloadFileAndUpload.从OSS中下载文件，并且上传文件到SFTP,下载和上传操作处于同步流程中，开始处理,集合总数：{}", batchList.size());
        for (Map<String, List<HuanengImageDataEntity>> stringListMap : batchList) {
            for (Map.Entry<String, List<HuanengImageDataEntity>> next : stringListMap.entrySet()) {
                String key = next.getKey();
                String iouKey = key.substring(0, key.lastIndexOf("_"));
                String iouId = key.substring(key.lastIndexOf("_") + 1);
                List<HuanengImageDataEntity> value = next.getValue();
                for (HuanengImageDataEntity entity : value) {
                    try {
                        // 以借据维度生成文件目录: /upload/morse/借据号
                        String uploadRemotePath = CommonSftpComponent.remoteFilePath + iouKey + SLASH;
                        String archiveUrl = entity.getArchiveUrl();
                        String fileName = entity.getArchiveFileName() + archiveUrl.substring(archiveUrl.lastIndexOf("."));
                        // 判断文件是否存在,false 表示不存在，则需要下载和上传操作
                        if (!existStaticFile(sftpUtil, uploadRemotePath, fileName)) {
                            // 下载操作
                            byte[] fileBuffer = null;//jrssSdkComponent.downloadFile(archiveUrl);

                            // 上传操作
                            InputStream inputStream = new ByteArrayInputStream(fileBuffer);
                            sftpUtil.uploadFile(inputStream, fileName, uploadRemotePath);
                        }
                    } catch (Exception e) {
                        sftpUtil = commonSftpComponent.initSftpConnection();
                        addData(iouId);
                        logger.error("DownloadFileAndUploadFile.从OSS下载文件,上传到SFTP服务失败，网贷借据id：{},贷款单号：{}", iouId, key, e);
                        logger.error("上传文件失败，失败借据id集合：{}", JSONObject.toJSONString(localQueue));
                    }
                }
            }
            logger.info("当前循环，下载文件，并且上传文件到SFTP服务器,处理成功");
        }
    }

    public synchronized void addData(String iouId) {
        int minStartId = findMin(localQueue);
        int newIouId = Integer.parseInt(iouId);
        if (minStartId > newIouId) {
            localQueue.add(newIouId);
        }
    }


    /**
     * @Description：上传文件到SFTP
     * @Author： ext.fengcheng
     * @Date：2023/9/19 17:22
     * @Param： [fileBufferMap]
     * @Return： void
     */
    public void uploadFileToSftp(Map<String, Map<String, byte[]>> fileBufferMap) {
        logger.info("uploadFileToSftp.上传文件到华能SFTP服务器,开始处理");
        fileBufferMap.forEach((key, value) -> value.forEach((fileNamekey, fileBuffer) -> {
            // 以借据维度生成文件目录: /upload/morse/借据号
            try {
                InputStream inputStream = new ByteArrayInputStream(fileBuffer);
                commonSftpComponent.sftpUtil.uploadFile(inputStream, fileNamekey, CommonSftpComponent.remoteFilePath + key + SLASH);
            } catch (Exception e) {
                logger.error("上传文件到华能SFTP服务器时发生异常，文件上传路径：{}, 异常信息：{}", CommonSftpComponent.remoteFilePath + key + SLASH, e.getMessage(), e);
                throw new RuntimeException(e);
            }
        }));
        logger.info("uploadFileToSftp.上传文件到华能SFTP服务器，处理结束");
    }

    /**
     * @Description：从SFTP下载文件到本地
     * @Author： ext.fengcheng
     * @Date：2023/9/27 16:47
     * @remotePath： SFTP文件目录
     * @remoteFileName： SFTP文件目录下的文件名
     * @localPath： 本地文件目录
     * @localFileName： 本地文件目录下存放文件名
     * @Return： boolean
     */
    @Override
    public boolean downloadFileToSftp(String remotePath, String remoteFileName, String localPath, String localFileName, boolean isBatchDownload) {
        try {
            if (ObjectUtils.isEmpty(commonSftpComponent.sftpUtil)) commonSftpComponent.initStaticSftpConnection();
            if (isBatchDownload) {
                logger.info("开始进行批量文件下载");
                commonSftpComponent.sftpUtil.batchDownLoadFile(remotePath, localPath);
            } else {
                logger.info("开始进行单个文件下载");
                commonSftpComponent.sftpUtil.downloadFile(remotePath, remoteFileName, localPath, localFileName);
            }
        } catch (Exception e) {
            logger.error("上传到华能SFTP,发生异常，异常信息：{}", e.getMessage(), e);
            throw new RuntimeException(e);
        } finally {
            commonSftpComponent.closeSftpConnection();
        }
        return true;
    }

    @Override
    public Integer countUploadsFileTotal() {
        int size = 0;
        SftpUtil sftpObj = commonSftpComponent.initSftpConnection();
        Vector listFiles = sftpObj.listFiles(CommonSftpComponent.remoteFilePath);
        if (ObjectUtils.isNotEmpty(listFiles)) {
            size = listFiles.size();
            logger.info("countUploadsFileTotal.已上传的借据总数为：【" + size + "】.最小借据id：【" + findMin(localQueue) + "】.失败借据id集合：{}", JSONObject.toJSONString(localQueue));
            sftpObj.disconnect();
        }
        return size;
    }

    @Override
    public Boolean clearLocalCache() {
        if (!CollectionUtils.isEmpty(localQueue)) {
            localQueue.clear();
        }
        return true;
    }

    /**
     * 拆分Map
     *
     * @param fileKeyMap
     * @return
     */
    public static List<Map<String, List<HuanengImageDataEntity>>> splitMap(Map<String, List<HuanengImageDataEntity>> fileKeyMap) {
        List<Map<String, List<HuanengImageDataEntity>>> list = new ArrayList<>();
        for (Map.Entry<String, List<HuanengImageDataEntity>> entry : fileKeyMap.entrySet()) {
            Map<String, List<HuanengImageDataEntity>> smallMap = new HashMap<>();
            smallMap.put(entry.getKey(), entry.getValue());
            list.add(smallMap);
        }
        return list;
    }

    /**
     * 获取最小值
     *
     * @param queue
     * @return
     */
    private static int findMin(ConcurrentLinkedQueue<Integer> queue) {
        int min = Integer.MAX_VALUE;
        for (int value : queue) {
            if (value < min) {
                min = value;
            }
        }
        return min;
    }
}
